package com.linguochao.rabbitmq.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Configuration
public class RabbitBatchConfig {

    @Value("${spring.rabbitmq.listener.batch.size:7}")
    private Integer batchSize;

    /**
     * 创建 RabbitAdmin 类，这个类封装了对 RabbitMQ 管理端的操作！
     *
     * @param connectionFactory
     * @return
     */
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        // 只有设置为 true，spring 才会加载 RabbitAdmin 这个类
        rabbitAdmin.setAutoStartup(true);

        // 声明交换机
        rabbitAdmin.declareExchange(new DirectExchange(BATCH_EXCHANGE_NAME));

        Map<String, Object> arguments = new HashMap<>();
        // 单节点活跃
        arguments.put("x-single-active-consumer", true);

        for (String queue : queueList) {
            // 声明队列
            rabbitAdmin.declareQueue(new Queue(queue,true,false,false,arguments));
            // 绑定
            rabbitAdmin.declareBinding(new Binding(queue,Binding.DestinationType.QUEUE,BATCH_EXCHANGE_NAME,queue,null));
        }

        return rabbitAdmin;
    }


    //  mq的容器工厂, 配置批处理
    @Bean("batchQueueRabbitListenerContainerFactory")
    public SimpleRabbitListenerContainerFactory batchQueueRabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        //设置批量
        factory.setBatchListener(true);
        factory.setConsumerBatchEnabled(true);//设置BatchMessageListener生效
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置手动ack
        factory.setBatchSize(batchSize);//设置监听器一次批量处理的消息数量
        factory.setDeBatchingEnabled(true);
        return factory;
    }

    //======================批量监听========================================


    // 批处理交换机
    public static final String BATCH_EXCHANGE_NAME = "batch.queue.exchange";

    // 批处理routing-key
    public static final String BATCH_QUEUE_ROUTING_KEY = "batch.queue.routing.key";

    // 声明批处理交换机
    @Bean(BATCH_QUEUE_ROUTING_KEY)
    public DirectExchange batchExchange(){
        return new DirectExchange(BATCH_EXCHANGE_NAME);
    }

    // 批处理消费队列
    public static final String BATCH_QUEUE_NAME = "batch.queue";

    // 单节点活跃消费者
    @Bean(BATCH_QUEUE_NAME)
    public Queue batchQueue() {
        Map<String, Object> arguments = new HashMap<>();
        // 死信队列交换机
        arguments.put( "x-dead-letter-exchange", BATCH_EXCHANGE_NAME);
        // 死信队列routing-key
        arguments.put("x-dead-letter-routing-key", BATCH_QUEUE_ROUTING_KEY);
        // 单节点活跃
        arguments.put("x-single-active-consumer", true);
        Queue queue = new Queue(BATCH_QUEUE_NAME, true, false, false, arguments);
        return queue;
    }

    // 声明批处理队列绑定关系
    @Bean
    public Binding batchLetterBindingA(@Qualifier(BATCH_QUEUE_NAME) Queue queue,
                                       @Qualifier(BATCH_QUEUE_ROUTING_KEY) DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(BATCH_QUEUE_ROUTING_KEY);
    }


    /**
     * 死信队列消息的超时时间：实现延迟投递
     */
    @Value("${spring.rabbitmq.listener.batch.ttlTime:60000}")
    private int xMessageTtlLong;

    // 批处理消费重试队列
    public static final String BATCH_RETRY_QUEUE_NAME = "batch.retry.queue";

    // 声明批处理重试队列
    @Bean
    public Queue batchRetryQueue() {
        Map<String, Object> args = new ConcurrentHashMap<>(3);
        // 死信队列交换机
        args.put( "x-dead-letter-exchange", BATCH_EXCHANGE_NAME);
        // 死信队列routing-key
        args.put("x-dead-letter-routing-key", BATCH_QUEUE_ROUTING_KEY);
        // 死信队列消息的超时时间
        args.put("x-message-ttl", xMessageTtlLong);
        return QueueBuilder.durable(BATCH_RETRY_QUEUE_NAME).withArguments(args).build();
    }

    // 批处理消费失败队列
    public static final String BATCH_FAIL_QUEUE_NAME = "batch.fail.queue";

    // 声明批处理消费失败队列
    @Bean(BATCH_FAIL_QUEUE_NAME)
    public Queue batchFailQueue() {
        return new Queue(BATCH_FAIL_QUEUE_NAME);
    }


    @Value("${spring.rabbitmq.batchQueue}")
    private String[] queueList;

    //======================批量监听========================================
}
